/*
 * REFIOEventService.cpp
 *
 *  Created on: May 29, 2013
 *      Author: Lucifer
 */

#include "../pch.h"
#include "REFIOEventContext.h"
#include "REFIOEventDispatcher.h"
#include "REFIOEventService.h"

#ifdef	BFX_IO_HAS_IOCP
#include <Windows.h>
#endif

namespace BFX
{
////////////////////////////////////////////////////////////////////////////////
// REFIOEventService

#ifdef	BFX_IO_HAS_IOCP

REFIOEventService::REFIOEventService(int nNumConcurrentThreads)
{
	BFX_ENSURE(nNumConcurrentThreads == -1 || (nNumConcurrentThreads >= 1 && nNumConcurrentThreads <= 256));

	m_lStopped = 0;
	m_lShutdown = 0;
	m_lNumOutstandingWorks = 0;

	m_nNumConcurrentThreads = nNumConcurrentThreads;
	if (m_nNumConcurrentThreads == -1)
		m_nNumConcurrentThreads = REFThread::GetNumberOfProcessors();

	// create a handle for the completion port.
	m_hCompletion = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, (DWORD) nNumConcurrentThreads);
	if (m_hCompletion == NULL)
	{
		DWORD dwLastError = ::GetLastError();
		BFX_DBGERROR_("IO")("Failed to initializes IO event service. error code: %08x.", dwLastError);
		throw SystemException(dwLastError, "Failed to initializes IO event service.");
	}
}

REFIOEventService::~REFIOEventService()
{
	Shutdown();
}

void REFIOEventService::Shutdown()
{
	// NOTE: after shutdown flag been set, no nceessary to consider concurrency anymore.
	::InterlockedExchange(&m_lShutdown, 1);

	while (InterlockedExchangeAdd(&m_lNumOutstandingWorks, 0) > 0)
	{
		bool bQueueEmpty = (m_completedContexts.GetSize() == 0);
		if (!bQueueEmpty)
		{
			// TODO: cancel all timers.

			// cancel all completed operations immediately.
			while (m_completedContexts.GetSize() > 0)
			{
				REF<REFIOEventContext> pContext = m_completedContexts.GetFirst();
				m_completedContexts.RemoveFirst();

				::InterlockedDecrement(&m_lNumOutstandingWorks);

				// NOTE: should we set the dwErrorCode to ERROR_OPERATION_ABORTED to indicates the I/O operation has been
				// aborted because of either a thread exit or an application request.
				pContext->Complete(0, 0);
			}
		}
		else
		{
			// cancel all waiting I/O completion handlers. 
			DWORD dwBytesTransferred = 0;
			ULONG_PTR lCompletionKey = 0;
			LPOVERLAPPED pOverlapped = NULL;
			::GetQueuedCompletionStatus(m_hCompletion, &dwBytesTransferred, &lCompletionKey, &pOverlapped,
				WAITING_MAX_TIMEOUT);
			if (pOverlapped)
			{
				::InterlockedDecrement(&m_lNumOutstandingWorks);

				// NOTE: should we set the dwErrorCode to ERROR_OPERATION_ABORTED to indicates the I/O operation has been
				// aborted because of either a thread exit or an application request.
				static_cast<REFIOEventContext*>(pOverlapped)->Complete(0, 0);
			}
		}
	}
}

// Request invocation of the given completion. Assumes that WorkStarted() was called previously.
void REFIOEventService::PostDeferredCompletion(REFIOEventContext* pContext)
{
	BFX_REQUIRE(pContext != NULL);

	pContext->m_lCompleted = 1;

	// enqueue the operation on the I/O completion port.
	if (!::PostQueuedCompletionStatus(m_hCompletion, 0, OVERLAPPED_CONTAINS_RESULT, pContext))
	{
		// out of resources. put on completed queue instead.
		MutexHolder holder(&m_mutex);

		m_completedContexts.AddLast(pContext);
		::InterlockedExchange(&m_lDispatchRequired, 1);
	}
}

// Run the event loop until Stop() be called, or no more works.
int REFIOEventService::Run()
{
	if (::InterlockedExchangeAdd(&m_lNumOutstandingWorks, 0) == 0)
	{
		// no more works to do.
		Stop();
		return 0;
	}

	int n = 0;
	while (RunOnce())
	{
		BFX_ASSERT(n >= 0);
		REFAutoreleasePool::Drain();

		// NOTE: we have to check it to prevent overflow. the remaining works will be running in next runloop.
		if ((++n) == INT_MAX)
			break;
	}

	return n;
}

/**
 * Dequeues at most one handler from the I/O completion port, and then executes it.
 * @return True if an operation that were dequeued.
 */
bool REFIOEventService::RunOnce()
{
	BFX_DBGTRACE_("IO")("run once.");

	for (;;)
	{
		// try to acquire responsibility for dispatching timers and completed operations.
		if (::InterlockedCompareExchange(&m_lDispatchRequired, 0, 1) == 1)
		{
			MutexHolder holder(&m_mutex);

			// TODO: dispatch pending timers and completed operations.

			// dispatch completed operations.
			while (m_completedContexts.GetSize() > 0)
			{
				REF<REFIOEventContext> pContext = m_completedContexts.GetFirst();
				m_completedContexts.RemoveFirst();
				PostDeferredCompletion(pContext);
			}
		}

		// get the next operation from the queue.
		DWORD dwBytesTransferred = 0;
		ULONG_PTR lCompletionKey = 0;
		LPOVERLAPPED pOverlapped = NULL;

		::SetLastError(NOERROR);
		BOOL bRetVal = GetQueuedCompletionStatus(m_hCompletion, &dwBytesTransferred, &lCompletionKey, &pOverlapped,
				WAITING_MAX_TIMEOUT);
		DWORD dwLastError = ::GetLastError();

		if (pOverlapped)
		{
			REFIOEventContext* pContext = static_cast<REFIOEventContext*>(pOverlapped);

	        // we may have been passed the nLastError and nBytesTransferred in the OVERLAPPED structure itself.
			if (lCompletionKey == OVERLAPPED_CONTAINS_RESULT)
			{
				dwLastError = pContext->Offset;
				dwBytesTransferred = pContext->OffsetHigh;
			}
			// otherwise ensure any result has been saved into the OVERLAPPED structure.
			else
			{
				pContext->Offset = dwLastError;
				pContext->OffsetHigh = dwBytesTransferred;
			}

			// dispatch the event only if ready.
			// the event may not be ready if the initiating function (e.g. call to WSARecv) has not yet returned.
			// this is because the initiating function still want access to the context's OVERLAPPED structure.
			if (::InterlockedCompareExchange(&pContext->m_lCompleted, 1, 0) == 1)
			{
				// This helper class to ensure the number of outstanding works is decremented on block exit.
				struct TaskBlockExitHolder
				{
					REFIOEventService* m_pService;
					~TaskBlockExitHolder()
					{
						m_pService->WorkFinished();
					}
				} finalizer = { this };
				(void) finalizer;

				pContext->Complete((int)dwLastError, (int)dwBytesTransferred);
				return true;
			}
		}
		// time-out occurs, or the completion port handle associated with it is closed.
		else if (!bRetVal)
		{
			if (dwLastError != WAIT_TIMEOUT)
			{
				BFX_DBGWARN_("IO")("failed to get queued completion status. error code: %08x", dwLastError);
				return false;
			}
			// If we're not polling we need to keep going until we get a real handler.
			continue;
		}
		else if (lCompletionKey == WAKE_FOR_DISPATCH)
		{
			// we have been woken up to try to acquire responsibility for dispatching timers and completed operations.
		}
		else
		{
			// the stopped flag is always checked to ensure that leftover interrupts from previous run invocation are
			// ignored.
			if (::InterlockedExchangeAdd(&m_lStopped, 0) != 0)
			{
				// wake up next thread that is blocked on GetQueuedCompletionStatus().
				if (!PostQueuedCompletionStatus(m_hCompletion, 0, 0, NULL))
				{
					dwLastError = ::GetLastError();
					BFX_DBGERROR_("IO")("failed to wake up next thread. error code: %08x", dwLastError);
					throw SystemException(dwLastError);
				}
				return false;
			}
		}
	}
}

// Request to stop the event processing loop.
void REFIOEventService::Stop()
{
	BFX_DBGTRACE_("IO")("stop the event processing loop.");

	if (::InterlockedExchange(&m_lStopped, 1) == 0)
	{
		if (!::PostQueuedCompletionStatus(m_hCompletion, 0, 0, NULL))
		{
			DWORD dwLastError = ::GetLastError();
			BFX_DBGERROR_("IO")("error on stop the event processing loop. error code: %08x", dwLastError);
			throw SystemException(dwLastError);
		}
	}
}

// Reset in preparation for a subsequent run invocation.
void REFIOEventService::Restart()
{
	BFX_DBGTRACE_("IO")("reset the event processing loop.");

	::InterlockedExchange(&m_lStopped, 0);
}

// Increases number of outstanding works.
void REFIOEventService::WorkStarted()
{
	::InterlockedIncrement(&m_lNumOutstandingWorks);

	BFX_DBGTRACE_("IO")("work started. count: %d", m_lNumOutstandingWorks);
}

// Decreases number of outstanding works.
void REFIOEventService::WorkFinished()
{
	if (::InterlockedDecrement(&m_lNumOutstandingWorks) == 0)
		Stop();

	BFX_DBGTRACE_("IO")("work finished. count: %d", m_lNumOutstandingWorks);
}

// This Method binds the Socket Win32 Handle to the service's CompletionPort (make sure we only bind once per socket).
void REFIOEventService::BindToCompletionPort(HANDLE handle)
{
	BFX_DBGTRACE_("IO")("binding socket handle #(%08x) to completion port.", handle);
	BFX_ASSERT(handle != NULL && handle != INVALID_HANDLE_VALUE);

	// associate the socket handle with the completion port.
	if (::CreateIoCompletionPort(handle, m_hCompletion, 0, 0) == NULL)
	{
		DWORD dwLastError = ::GetLastError();
		BFX_DBGERROR_("IO")("failed to bind socket handle #(%08x) to completion port. error code: %08x.", handle,
				dwLastError);
		throw SystemException(dwLastError);
	}
}

void REFIOEventService::OnPending(REFIOEventContext* pContext)
{
	BFX_DBGTRACE_("IO")("on pending (%08p).", pContext);

	WorkStarted();

	if (::InterlockedCompareExchange(&pContext->m_lCompleted, 1, 0) == 1)
	{
		// enqueue the operation on the I/O completion port.
		if (!::PostQueuedCompletionStatus(m_hCompletion, 0, OVERLAPPED_CONTAINS_RESULT, pContext))
		{
			BFX_DBGWARN_("IO")("out of completion I/O resources, put on completed queue instead.");

			// out of resources. put on completed queue instead.
			MutexHolder holder(&m_mutex);

			m_completedContexts.AddLast(pContext);
			InterlockedExchange(&m_lDispatchRequired, 1);
		}
	}
}

void REFIOEventService::OnCompletion(REFIOEventContext* pContext, DWORD dwLastError, DWORD dwBytesTransferred)
{
	BFX_DBGTRACE_("IO")("on completion (%08p).", pContext);

	WorkStarted();

	// Flag that the operation is ready for invocation.
	pContext->m_lCompleted = 1;

	// store results in the OVERLAPPED structure.
	pContext->Offset = dwLastError;
	pContext->OffsetHigh = dwBytesTransferred;

	// enqueue the operation on the I/O completion port.
	if (!::PostQueuedCompletionStatus(m_hCompletion, 0, OVERLAPPED_CONTAINS_RESULT, pContext))
	{
		BFX_DBGWARN_("IO")("out of completion I/O resources, put on completed queue instead.");

		// out of resources. put on completed queue instead.
		MutexHolder holder(&m_mutex);

		m_completedContexts.AddLast(pContext);
		InterlockedExchange(&m_lDispatchRequired, 1);
	}
}

#else	//	BFX_IO_HAS_IOCP

REFIOEventService::REFIOEventService(int nNumConcurrentThreads)
{
	BFX_ENSURE(nNumConcurrentThreads == -1 || (nNumConcurrentThreads >= 1 && nNumConcurrentThreads <= 256));

	m_bShutdown = false;
	m_bStopped = false;
	m_lNumOutstandingWorks = 0;

	m_nNumConcurrentThreads = nNumConcurrentThreads;
	if (m_nNumConcurrentThreads == -1)
		m_nNumConcurrentThreads = REFThread::GetNumberOfProcessors();

	// This dummy context uses to enter event dispatching.
	struct DummyContext : public REFIOEventContext {
		DummyContext() : REFIOEventContext(IOEventCallback()) {}
		virtual bool Perform() { return true; };
	};
	m_pEventDispatchContext = new DummyContext();

	m_bDispatching = false;
	m_pEventDispatcher = new REFIOEventDispatcher(this);
}

REFIOEventService::~REFIOEventService()
{
	Shutdown();
}

void REFIOEventService::Shutdown()
{
	SyncBlock::SyncLock lock(&m_syncBlock);
	m_bShutdown = true;
	lock.Release();

	while (m_completedContexts.GetSize() > 0)
	{
		REF<REFIOEventContext> pContext = m_completedContexts.GetFirst();
		m_completedContexts.RemoveFirst();
		if (pContext != m_pEventDispatchContext)
		{
			// TODO: set error code to indicates operation aborted.
			pContext->Complete();
		}
	}
}

REFIOEventDispatcher* REFIOEventService::GetEventDispatcher() const
{
	return m_pEventDispatcher;
}

void REFIOEventService::PostImmediateCompletion(REFIOEventContext* pContext)
{
	BFX_DBGTRACE_("IO")("post immediate completion (%08x)...", pContext);

	WorkStarted();

	PostDeferredCompletion(pContext);
}

// Request invocation of the given completion. Assumes that WorkStarted() was called previously.
void REFIOEventService::PostDeferredCompletion(REFIOEventContext* pContext)
{
	SyncBlock::SyncLock lock(&m_syncBlock);

	m_completedContexts.AddLast(pContext);
	m_syncBlock.Signal();
}

// Initialize the event dispatcher, if required.
void REFIOEventService::InitializeEventDispatcher()
{
	BFX_DBGTRACE_("IO")("Initializing event dispatcher.");

	if (!m_bShutdown)
	{
		// NOTE: the dummy context will not actually execute, just notify the service thread to enter the
		// event dispatch progress.
		PostDeferredCompletion(m_pEventDispatchContext);
	}
}

// Run the event loop until stopped or no more works.
int REFIOEventService::Run()
{
	if (m_lNumOutstandingWorks == 0)
	{
		Stop();
		return 0;
	}

	int n = 0;
	while (RunOnce())
	{
		BFX_ASSERT(n >= 0);
		REFAutoreleasePool::Drain();

		// NOTE: we have to check it to prevent overflow. the remaining works will be running in next runloop.
		if ((++n) == INT_MAX)
			break;
	}
	return n;
}

bool REFIOEventService::RunOnce()
{
	BFX_DBGTRACE_("IO")("run once.");

	while (!m_bStopped)
	{
		REF<REFIOEventContext> pContext;
		bool bMoreHandlers = false;

		SyncBlock::SyncLock lock (&m_syncBlock);
		if (m_completedContexts.GetSize() > 0)
		{
			// prepare to execute first handler from queue.
			pContext = m_completedContexts.GetFirst();
			m_completedContexts.RemoveFirst();
			bMoreHandlers = (m_completedContexts.GetSize() > 0);
		}
		else
		{
			m_syncBlock.Wait();
		}
		lock.Release();

		if (pContext == NULL)
		{
			continue;
		}
		else if (pContext == m_pEventDispatchContext)
		{
			m_bDispatching = !bMoreHandlers;

			REF<REFIOEventContextQueue> pCompletedContexts = new REFIOEventContextQueue();

			// This class uses to perform finalize operations on block exit. even if the DispatchEvents() raises an
			// exception.
			struct TaskBlockExitHolder
			{
				REFIOEventService* m_pService;
				REFIOEventContextQueue*	m_pContexts;
				~TaskBlockExitHolder()
				{
					// enqueue the completed contexts and reinsert the event dispatch context at the end of
					// the operation queue.
					m_pService->m_bDispatching = false;
					while (!m_pContexts->IsEmpty())
					{
						REF<REFIOEventContext> pContext = m_pContexts->Dequeue();
						m_pService->PostDeferredCompletion(pContext);
					}
					m_pService->InitializeEventDispatcher();
				}
			} holder = { this, pCompletedContexts };
			(void) holder;

			// enter the event dispatching, only block if the context queue is empty, otherwise we want to return as
			// soon as possible.
			// NOTE: May throw an exception.
			m_pEventDispatcher->DispatchEvents(!bMoreHandlers, pCompletedContexts);
		}
		else
		{
			// This class uses to perform finalize operations on block exit.
			struct BlockExitHolder
			{
				REFIOEventService* m_pService;
				~BlockExitHolder()
				{
					m_pService->WorkFinished();
				}
			} holder = { this };
			(void) holder;

			// complete the operation.
			// NOTE: May throw an exception.
			pContext->Complete();

			return true;
		}
	}
	return false;
}

// Request to stop the event processing loop.
void REFIOEventService::Stop()
{
	BFX_DBGTRACE_("IO")("stop the event processing loop.");

	SyncBlock::SyncLock lock (&m_syncBlock);

	m_bStopped = true;
	m_syncBlock.SignalAll();

	if (m_bDispatching)
	{
		m_bDispatching = false;
		m_pEventDispatcher->Interrupt();
	}
}

// Reset in preparation for a subsequent run invocation.
void REFIOEventService::Restart()
{
	BFX_DBGTRACE_("IO")("reset the event processing loop.");

	SyncBlock::SyncLock lock(&m_syncBlock);
	m_bStopped = false;
}

// Increases number of outstanding works.
void REFIOEventService::WorkStarted()
{
	Interlocked::Increment(&m_lNumOutstandingWorks);

	BFX_DBGTRACE_("IO")("work started. count: %d", m_lNumOutstandingWorks);
}

// Decreases number of outstanding works.
void REFIOEventService::WorkFinished()
{
	if (Interlocked::Decrement(&m_lNumOutstandingWorks) == 0)
		Stop();

	BFX_DBGTRACE_("IO")("work finished. count: %d", m_lNumOutstandingWorks);
}

#endif	//	BFX_IO_HAS_IOCP

} /* namespace BFX */
